工具与集成（Tools and Integrations）

概述

CrewAI 提供了丰富的工具生态系统，让智能体能够与外部世界交互。工具是智能体执行具体操作的能力扩展，包括搜索、文件操作、数据库查询等。

安装工具包

# 安装基础工具
pip install 'crewai[tools]'

# 安装 MCP 支持
pip install 'crewai-tools[mcp]'

内置工具

工具分类

类别	工具	说明
文件管理	FileReadTool, FileWriteTool	读写文件
网页爬取	ScrapeWebsiteTool, SeleniumScrapingTool	网页数据抓取
搜索	SerperDevTool, BraveSearchTool, TavilySearchTool	网络搜索
数据库	MySQLSearchTool, SnowflakeSearchTool	数据库查询
向量数据库	QdrantVectorSearchTool, WeaviateVectorSearchTool	向量搜索
AI 能力	DallETool, VisionTool, CodeInterpreterTool	AI 增强功能
文档处理	PDFSearchTool, DOCXSearchTool, CSVSearchTool	文档搜索

常用工具示例

SerperDevTool（搜索）

from crewai_tools import SerperDevTool

# 需要设置环境变量 SERPER_API_KEY
search_tool = SerperDevTool()

agent = Agent(
    role="Researcher",
    goal="Find relevant information",
    tools=[search_tool]
)

FileReadTool（文件读取）

from crewai_tools import FileReadTool

file_tool = FileReadTool()

agent = Agent(
    role="Document Analyst",
    goal="Analyze documents",
    tools=[file_tool]
)

ScrapeWebsiteTool（网页爬取）

from crewai_tools import ScrapeWebsiteTool

scraper = ScrapeWebsiteTool()

agent = Agent(
    role="Web Researcher",
    goal="Extract web content",
    tools=[scraper]
)

CodeInterpreterTool（代码执行）

from crewai_tools import CodeInterpreterTool

code_tool = CodeInterpreterTool()

agent = Agent(
    role="Developer",
    goal="Execute code",
    tools=[code_tool]
)

自定义工具

方式 1：继承 BaseTool

from typing import Any, Type
from pydantic import BaseModel, Field
from crewai.tools import BaseTool

# 定义输入 Schema
class WeatherInput(BaseModel):
    """Input schema for WeatherTool."""
    city: str = Field(..., description="City name")
    country: str = Field(None, description="Country code (optional)")

class WeatherTool(BaseTool):
    name: str = "Weather Search"
    description: str = "Get current weather for a city"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str, country: str = None) -> str:
        """Execute the tool."""
        # 实现逻辑
        import requests
        api_key = os.environ.get("WEATHER_API_KEY")
        q = f"{city},{country}" if country else city
        response = requests.get(
            f"https://api.openweathermap.org/data/2.5/weather",
            params={"q": q, "appid": api_key}
        )
        data = response.json()
        return f"Weather in {city}: {data['weather'][0]['main']}"

# 使用自定义工具
weather_tool = WeatherTool()
agent = Agent(role="Weather Reporter", tools=[weather_tool])

方式 2：使用 @tool 装饰器

from crewai import tool

@tool("Calculate")
def calculate(expression: str) -> str:
    """Calculate a mathematical expression.

    Args:
        expression: The math expression to evaluate
    """
    try:
        result = eval(expression)
        return f"Result: {result}"
    except Exception as e:
        return f"Error: {e}"

# 使用装饰器创建的工具
agent = Agent(
    role="Calculator",
    goal="Perform calculations",
    tools=[calculate]
)

完整自定义工具示例

from typing import Any, List, Type
import os
from pydantic import BaseModel, Field
from crewai.tools import BaseTool, EnvVar

class GitHubSearchInput(BaseModel):
    """Input for GitHub search."""
    query: str = Field(..., description="Search query")
    language: str = Field(None, description="Programming language filter")
    max_results: int = Field(5, ge=1, le=50, description="Max results")

class GitHubSearchTool(BaseTool):
    name: str = "GitHub Search"
    description: str = "Search GitHub repositories"
    args_schema: Type[BaseModel] = GitHubSearchInput

    # 声明环境变量依赖
    env_vars: List[EnvVar] = [
        EnvVar(
            name="GITHUB_TOKEN",
            description="GitHub API token",
            required=True
        )
    ]

    # 声明包依赖
    package_dependencies: List[str] = ["PyGithub"]

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        # 验证环境变量
        if "GITHUB_TOKEN" not in os.environ:
            raise ValueError("GITHUB_TOKEN is required")

        # 惰性导入
        try:
            from github import Github
            self.client = Github(os.environ["GITHUB_TOKEN"])
        except ImportError:
            raise ImportError("Install PyGithub: pip install PyGithub")

    def _run(
        self,
        query: str,
        language: str = None,
        max_results: int = 5
    ) -> str:
        """Search GitHub repositories."""
        try:
            search_query = query
            if language:
                search_query += f" language:{language}"

            repos = self.client.search_repositories(
                query=search_query,
                sort="stars",
                order="desc"
            )

            results = []
            for i, repo in enumerate(repos[:max_results]):
                results.append(f"- {repo.full_name}: {repo.stargazers_count} stars")

            return "\n".join(results) if results else "No results found"
        except Exception as e:
            return f"Error searching GitHub: {e}"

    async def _arun(self, *args: Any, **kwargs: Any) -> str:
        """Async execution (delegates to sync)."""
        return self._run(*args, **kwargs)

MCP（Model Context Protocol）集成

CrewAI 支持 MCP 协议，可以使用社区提供的数千个工具。

方式 1：托管连接（推荐）

from mcp import StdioServerParameters
from crewai_tools import MCPServerAdapter
from crewai import Agent, Task, Crew

# STDIO 服务器
serverparams = StdioServerParameters(
    command="uvx",
    args=["--quiet", "pubmedmcp@0.1.3"],
    env={"UV_PYTHON": "3.12", **os.environ}
)

# 使用 with 语句自动管理连接
with MCPServerAdapter(serverparams) as tools:
    agent = Agent(
        role="Medical Researcher",
        goal="Search medical literature",
        tools=tools  # MCP 工具列表
    )

    task = Task(
        description="Search for recent papers on {topic}",
        expected_output="Summary of findings",
        agent=agent
    )

    crew = Crew(agents=[agent], tasks=[task])
    result = crew.kickoff(inputs={"topic": "COVID-19"})

方式 2：手动管理连接

from mcp import StdioServerParameters
from crewai_tools import MCPServerAdapter

serverparams = StdioServerParameters(
    command="uvx",
    args=["--quiet", "some-mcp-server"],
)

try:
    mcp_adapter = MCPServerAdapter(serverparams)
    tools = mcp_adapter.tools

    agent = Agent(role="Agent", tools=tools)
    # ... 使用 agent

finally:
    mcp_adapter.stop()  # 重要：确保关闭连接

SSE 服务器连接

serverparams = {"url": "http://localhost:8000/sse"}

with MCPServerAdapter(serverparams) as tools:
    agent = Agent(role="Agent", tools=tools)
    # ...

RAG 工具

CrewAI 提供 RAG（检索增强生成）工具：

使用 RagTool

from crewai_tools import RagTool

# 创建 RAG 工具
rag_tool = RagTool()

# 添加文档
rag_tool.add("path/to/document.pdf")
rag_tool.add("path/to/another.txt")

# 分配给智能体
agent = Agent(
    role="Knowledge Assistant",
    goal="Answer questions from documents",
    tools=[rag_tool]
)

自定义 RAG Adapter

from crewai_tools.tools.rag.rag_tool import Adapter, RagTool

class CustomAdapter(Adapter):
    store: list[str] = []

    def add(self, text: str, **kwargs) -> None:
        self.store.append(text)

    def query(self, question: str) -> str:
        # 自定义检索逻辑
        relevant = [t for t in self.store if question.lower() in t.lower()]
        return "\n".join(relevant) if relevant else "No relevant content"

class CustomRagTool(RagTool):
    name: str = "Custom RAG"
    description: str = "Custom knowledge retrieval"
    adapter: Adapter = CustomAdapter()

LLM 连接

CrewAI 支持多种 LLM 提供商：

OpenAI（默认）

# 设置环境变量
export OPENAI_API_KEY=sk-...

# 使用默认 OpenAI
agent = Agent(role="Agent", goal="...")

其他提供商

from crewai import Agent, LLM

# Anthropic Claude
agent = Agent(
    role="Agent",
    llm=LLM(
        model="claude-3-sonnet-20240229",
        api_key=os.environ["ANTHROPIC_API_KEY"]
    )
)

# Azure OpenAI
agent = Agent(
    role="Agent",
    llm=LLM(
        model="azure/gpt-4",
        api_key=os.environ["AZURE_API_KEY"],
        base_url=os.environ["AZURE_ENDPOINT"]
    )
)

# Ollama（本地）
agent = Agent(
    role="Agent",
    llm=LLM(
        model="ollama/llama2",
        base_url="http://localhost:11434"
    )
)

工具最佳实践

1. 清晰的描述

class MyTool(BaseTool):
    name: str = "Specific Tool Name"
    description: str = """
    What: Specifically what this tool does
    When: When to use this tool
    Input: What input it expects
    Output: What it returns
    """

2. 输入验证

class ToolInput(BaseModel):
    query: str = Field(..., min_length=1, description="Search query")
    limit: int = Field(10, ge=1, le=100, description="Result limit")

class MyTool(BaseTool):
    args_schema: Type[BaseModel] = ToolInput

3. 错误处理

def _run(self, query: str) -> str:
    try:
        result = self.api_call(query)
        return result
    except requests.Timeout:
        return "Request timed out. Please try again."
    except Exception as e:
        return f"Error: {str(e)}"

4. 环境变量管理

class MyTool(BaseTool):
    env_vars: List[EnvVar] = [
        EnvVar(name="API_KEY", required=True),
        EnvVar(name="API_URL", required=False, default="https://api.example.com")
    ]

与其他框架对比

特性	CrewAI	LangChain	MetaGPT
工具定义	BaseTool / @tool	@tool / BaseTool	Tool 基类
MCP 支持	原生支持	需要适配	不支持
输入验证	Pydantic	Pydantic	自定义
异步支持	_arun 方法	_arun 方法	async run
工具数量	80+ 内置	100+ 内置	较少

---

下一节：17.4 Flows